package com.shujia.flink.sink

import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.scala._

object Demo2SInkFunction {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //使用自定义sink
    linesDS.addSink(new MysqlSink)

    env.execute()
  }

}

class MysqlSink extends SinkFunction[String] {
  /**
   * invoke： 用于将数据sink出去的方法，一条数据执行一次
   *
   * 可以自定义sink到任何存储中
   *
   * @param value   ： 一条数据
   * @param context ： flink上下游文件
   */
  override def invoke(value: String, context: SinkFunction.Context): Unit = {

    println(s"mySInk:$value")

  }
}
